package com.example.rxjava2.activity;

import android.support.v7.app.AppCompatActivity;
import android.os.Bundle;
import android.util.Log;
import android.view.View;

import com.example.rxjava2.R;

import io.reactivex.Observable;
import io.reactivex.ObservableEmitter;
import io.reactivex.ObservableOnSubscribe;
import io.reactivex.android.schedulers.AndroidSchedulers;
import io.reactivex.functions.Consumer;
import io.reactivex.functions.Function;
import io.reactivex.schedulers.Schedulers;

public class ThreadActivity extends AppCompatActivity {

    private static final String TAG = ThreadActivity.class.getSimpleName();

    @Override
    protected void onCreate(Bundle savedInstanceState) {
        super.onCreate(savedInstanceState);
        setContentView(R.layout.activity_thread);
        findViewById(R.id.btn_thread).setOnClickListener(new View.OnClickListener() {
            @Override
            public void onClick(View v) {
                testThread();
            }
        });

        findViewById(R.id.btn_thread1).setOnClickListener(new View.OnClickListener() {
            @Override
            public void onClick(View v) {
                testThread1();
            }
        });

        findViewById(R.id.btn_thread2).setOnClickListener(new View.OnClickListener() {
            @Override
            public void onClick(View v) {
                testThread2();
            }
        });
        findViewById(R.id.btn_thread3).setOnClickListener(new View.OnClickListener() {
            @Override
            public void onClick(View v) {
                testThread3();
            }
        });
    }

    private void testThread3() {
        Observable.create(new ObservableOnSubscribe<Integer>() {
            @Override
            public void subscribe(ObservableEmitter<Integer> e) throws Exception {
                for (int i = 0; i < 5; i++) {
                    Log.i(TAG, "subscribe: " + Thread.currentThread().getName() + "发射 ：" + i);
                    e.onNext(i);
                    Thread.sleep(1000);
                }
            }
        }).subscribeOn(Schedulers.io())//1 cachethrea
                .observeOn(Schedulers.newThread())
                .map(new Function<Integer, Integer>() {
                    @Override
                    public Integer apply(Integer integer) throws Exception {
                        Log.i(TAG, "apply: " + "接收" + Thread.currentThread().getName() + integer);
                        return integer + 10000;
                    }
                })
                .observeOn(AndroidSchedulers.mainThread())
                .subscribe(new Consumer<Integer>() {
                    @Override
                    public void accept(Integer integer) throws Exception {
                        Log.i(TAG, "accept: " + Thread.currentThread().getName() + "接收" + integer);
                    }
                });
    }

    private void testThread2() {
        Observable.create(new ObservableOnSubscribe<Integer>() {
            @Override
            public void subscribe(ObservableEmitter<Integer> e) throws Exception {
                for (int i = 0; i < 5; i++) {
                    Log.i(TAG, "subscribe: " + Thread.currentThread().getName() + "发射 ：" + i);
                    e.onNext(i);
                    Thread.sleep(1000);
                }
            }
        }).subscribeOn(Schedulers.io())//1 cachethread
                .map(new Function<Integer, Integer>() {
                    @Override
                    public Integer apply(Integer integer) throws Exception {
                        Log.i(TAG, "apply: " + Thread.currentThread().getName() + "处理" + integer);
                        return integer + 100000;
                    }
                })
                .subscribeOn(Schedulers.newThread())//2 newthread
                .observeOn(AndroidSchedulers.mainThread())
                .subscribe(new Consumer<Integer>() {
                    @Override
                    public void accept(Integer integer) throws Exception {
                        Log.i(TAG, "accept: " + Thread.currentThread().getName() + "接收" + integer);
                    }
                });
    }

    private void testThread() {
        new Thread(new Runnable() {
            @Override
            public void run() {
                Log.i(TAG, "run:-------> " + Thread.currentThread().getName());
                Observable.create(new ObservableOnSubscribe<Integer>() {
                    @Override
                    public void subscribe(ObservableEmitter<Integer> e) throws Exception {
                        for (int i = 0; i < 5; i++) {
                            Log.i(TAG, "subscribe: " + Thread.currentThread().getName() + "发射 ：" + i);
                            e.onNext(i);
                            Thread.sleep(1000);
                        }
                    }
                }).map(new Function<Integer, Integer>() {
                    @Override
                    public Integer apply(Integer integer) throws Exception {
                        return integer + 100;
                    }
                })
                        .subscribeOn(Schedulers.io())//设置被观察者在io线程
                        .observeOn(AndroidSchedulers.mainThread())//设置观察者在主线程处理
                        .subscribe(new Consumer<Integer>() {
                            @Override
                            public void accept(Integer integer) throws Exception {
                                Log.i(TAG, "accept: " + Thread.currentThread().getName() + "接收 ：" + integer);
                            }
                        });
            }
        }).start();

    }

    private void testThread1() {
        Observable.create(new ObservableOnSubscribe<Integer>() {
            @Override
            public void subscribe(ObservableEmitter<Integer> e) throws Exception {
                for (int i = 0; i < 5; i++) {
                    Log.i(TAG, "subscribe: " + Thread.currentThread().getName() + "发射 ：" + i);
                    e.onNext(i);
                    Thread.sleep(1000);
                }
            }
        }).subscribeOn(Schedulers.newThread())
                .map(new Function<Integer, Integer>() {
                    @Override
                    public Integer apply(Integer integer) throws Exception {
                        Log.i(TAG, "apply: " + Thread.currentThread().getName() + "处理" + integer);
                        return integer + 100000;
                    }
                })
                .observeOn(AndroidSchedulers.mainThread())
                .subscribe(new Consumer<Integer>() {
                    @Override
                    public void accept(Integer integer) throws Exception {
                        Log.i(TAG, "accept: " + Thread.currentThread().getName() + "接收" + integer);
                    }
                });
    }
}
